package rocketmq

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/injoyai/goutil/string/logs/v1"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"github.com/apache/rocketmq-client-go/v2/rlog"
	"log"
	"time"
)

type Config struct {
	// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
	Endpoint string
	// AccessKey ID阿里云身份验证，在阿里云RAM控制台创建。
	AccessKey string
	// AccessKey Secret阿里云身份验证，在阿里云RAM控制台创建。
	SecretKey string
	// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
	InstanceId string
	//调试
	Debug bool
}

type Entity struct {
	mqCfg    Config            //mq配置
	funcLog  func(string)      //日志
	producer rocketmq.Producer //生产者
}

func New(cfg Config) (*Entity, error) {
	rlog.SetLogger(logger{})
	p, err := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{cfg.Endpoint})),
		producer.WithNamespace(cfg.InstanceId),
		producer.WithInstanceName(cfg.InstanceId),
		producer.WithRetry(2),
		producer.WithCredentials(primitive.Credentials{
			AccessKey: cfg.AccessKey,
			SecretKey: cfg.SecretKey,
		}),
	)
	if err != nil {
		return nil, err
	}
	if err := p.Start(); err != nil {
		return nil, err
	}
	return &Entity{
		mqCfg:    cfg,
		producer: p,
	}, nil
}

func (this *Entity) SetConfig(cfg Config) *Entity {
	this.mqCfg = cfg
	return this
}

func (this *Entity) SetFunLog(fn func(string)) *Entity {
	this.funcLog = fn
	return this
}

func (this *Entity) Pub(topic string, msg string) error {
	//加到通讯日志,文件
	if this.funcLog != nil {
		this.funcLog(fmt.Sprintf("%s[%s][%s] : %s\n", time.Now().Format("2006-01-02 15:04:05 "), "发布", topic, msg))
	}
	if this.mqCfg.Debug {
		log.Println("[发布]["+topic+"] : ", msg)
	}
	result, err := this.producer.SendSync(context.Background(),
		primitive.NewMessage(topic, []byte(msg)))
	if err != nil {
		return err
	}
	if result.Status != primitive.SendOK {
		switch result.Status {
		case primitive.SendFlushDiskTimeout:
			return errors.New("超时:Disk")
		case primitive.SendFlushSlaveTimeout:
			return errors.New("超时:Slave")
		case primitive.SendSlaveNotAvailable:
			return errors.New("未激活")
		case primitive.SendUnknownError:
			return errors.New("未知错误")
		default:
			return errors.New("发布失败")
		}
	}
	return nil
}

func (this *Entity) Sub(topic, groupId string, fun func([]byte) error) {
	c, err := rocketmq.NewPushConsumer(
		consumer.WithGroupName(groupId),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{this.mqCfg.Endpoint})),
		consumer.WithNamespace(this.mqCfg.InstanceId),
		consumer.WithInstance(this.mqCfg.InstanceId),
		consumer.WithCredentials(primitive.Credentials{
			AccessKey: this.mqCfg.AccessKey,
			SecretKey: this.mqCfg.SecretKey,
		}),
	)
	if err != nil {
		logs.Err(err)
		return
	}

	logs.PrintErr(c.Subscribe(topic, consumer.MessageSelector{Expression: "*"},
		func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
			for _, v := range msgs {
				//加到通讯日志,文件
				if this.funcLog != nil {
					this.funcLog(fmt.Sprintf("%s[%s][%s] : %s\n", time.Now().Format("2006-01-02 15:04:05 "), "订阅", topic, string(v.Body)))
				}
				if this.mqCfg.Debug {
					log.Println("[订阅]["+topic+"] : ", string(v.Body))
				}
				logs.PrintErr(fun(v.Body))
			}
			//返回consumer.Rollback会重发
			return consumer.ConsumeSuccess, nil
		}))
	if err := c.Start(); err != nil {
		logs.Err(err)
	} else {
		log.Println("MQ(TCP)订阅成功:", topic)
	}
}
